package com.zhny.test;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;

import java.util.Arrays;
import java.util.List;

import org.apache.parquet.Strings;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.fpm.FPGrowth;
import org.apache.spark.mllib.fpm.FPGrowth.FreqItemset;
import org.apache.spark.mllib.fpm.FPGrowthModel;
import org.apache.spark.rdd.RDD;
//协同过滤
public class FPGrowthUtil {
    public static void exc(String dataFilePath, String resultFilePath) {
        SparkConf sparkConf = new SparkConf().setAppName("LinearRegressionTest").setMaster("local[1]");
        sparkConf.set("spark.driver.allowMultipleContexts", "true");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        JavaRDD<String> data = null;

        if (Strings.isNullOrEmpty(dataFilePath)) {
            data = sc.textFile("src/main/resources/data/FpGrowhData.txt");
        } else {
            data = sc.textFile(dataFilePath);
        }

        JavaRDD<List<String>> parsedData = data.map(new Function<String, List<String>>() {
            private static final long serialVersionUID = 1L;
            @Override
            public List<String> call(String line) throws Exception {
                String[] parts = line.split(" ");

                return Arrays.asList(parts);
            }
        }).cache();

        FPGrowthModel<String> fPGrowthModel = new FPGrowth().setNumPartitions(10).setMinSupport(0.25).run(parsedData);

        FileWriter fos = null;
        try {
            fos = new FileWriter(new File(resultFilePath));
            fos.write(fPGrowthModel.toString() + "\n");

            RDD<FreqItemset<String>> rdd = fPGrowthModel.freqItemsets();
            JavaRDD<FreqItemset<String>> javaRdd = new JavaRDD<FreqItemset<String>>(rdd, rdd.elementClassTag());
            List<FreqItemset<String>> result = javaRdd.collect();

            for (FreqItemset<String> item : result) {
                StringBuilder builder = new StringBuilder();

                for (String s : item.javaItems()) {
                    builder.append(s + ",");
                }

                builder.append(item.freq() + "\n");

                fos.write(builder.toString());
            }

            fos.flush();
            fos.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
